#include "FIFOSafeQueueTestSuite.h"
#include <bits/types/FILE.h>
#include <cstdint>
#include <cstdio>
#include <gtest/gtest.h>
#include <thread>
#include "FIFOQueue.h"
#include "XnccType.h"
#include "xncc_random.hpp"

TestQueue* FIFOSafeQueueTestSuite::queue = nullptr;

TEST_F(FIFOSafeQueueTestSuite, one_thread_write_read)
{
#ifdef DEBUG_PROD

    xncc::tools::xncc_random<xncc::types::queue_size_t> random(24, 500);
    xncc::tools::xncc_random<std::string>               randomStr;
    for (auto i = 0; i < 1000000; i++) {
        const auto len = random.get_value();
        const auto buf = randomStr.get_value(int(len));
        queue->write(buf.c_str(), len);
        queue->writeOverNotNotify(len);
        auto ptr = queue->read(len);
        if (ptr == nullptr) {
            std::cout << "nullptr i = " << i << " len " << len << " dump:" << queue->dump() << std::endl;
            return;
        }
        std::string read_data(ptr, len);
        queue->readOver(len);
        EXPECT_TRUE(read_data == buf);
        EXPECT_TRUE(ptr != nullptr);
    }
#endif
}
static constexpr uint32_t THREAD_MSG_HEAD_LEN = sizeof(xncc::msg::MsgHeaderInThread);

TEST_F(FIFOSafeQueueTestSuite, writeonce_read_twice)
{
#ifdef DEBUG_PROD

    bool                                                ouput_debug = false;
    xncc::tools::xncc_random<xncc::types::queue_size_t> random(24, 500);
    xncc::tools::xncc_random<std::string>               randomStr;
    std::cout << "0 queue dump:" << queue->dump() << std::endl;
    for (auto i = 0; i < 1000000; i++) {
        const auto len = random.get_value();
        const auto buf = randomStr.get_value(int(len));

        char                          buffer[THREAD_MSG_HEAD_LEN + 501] = {0};
        xncc::msg::MsgHeaderInThread* hptr = reinterpret_cast<xncc::msg::MsgHeaderInThread*>(buffer);
        hptr->len                          = static_cast<uint32_t>(len);
        hptr->id                           = 1000;
        hptr->connId                       = 1;
        hptr->tid                          = 1;
        memcpy(buffer + THREAD_MSG_HEAD_LEN, buf.c_str(), static_cast<uint32_t>(len));
        const xncc::types::queue_size_t total_len = THREAD_MSG_HEAD_LEN + len;
        if (ouput_debug) {
            std::cout << "index:" << i << " data_len:" << total_len << " 1 queue dump:" << queue->dump() << std::endl;
        }
        queue->write(buffer, total_len);
        queue->writeOverNotNotify(total_len);
        if (ouput_debug) {
            std::cout << "index:" << i << " data_len:" << total_len << " 2 queue dump:" << queue->dump() << std::endl;
        }

        char* readBuf = queue->read(THREAD_MSG_HEAD_LEN);
        if (unlikely(readBuf == nullptr)) {
            std::cout << "nullptr i = " << i << " len " << total_len << " dump:" << queue->dump() << std::endl;
        }
        xncc::msg::MsgHeaderInThread* head = reinterpret_cast<xncc::msg::MsgHeaderInThread*>(readBuf);
        if (head->len != len) {
            std::cout << "read error i = " << i << " len " << total_len << " dump:" << queue->dump() << std::endl;
        }
        EXPECT_EQ(head->len, len);
        EXPECT_EQ(head->id, 1000);
        EXPECT_EQ(head->connId, 1);
        EXPECT_EQ(head->tid, 1);

        if (ouput_debug) {
            std::cout << "index:" << i << " data_len:" << total_len << " 3 queue dump:" << queue->dump() << std::endl;
        }

        char* dataBuf = queue->read(static_cast<xncc::types::queue_size_t>(head->len) + THREAD_MSG_HEAD_LEN);
        EXPECT_TRUE(dataBuf != nullptr);
        if (likely(dataBuf)) {
            const auto data = std::string(dataBuf + THREAD_MSG_HEAD_LEN, head->len);
            EXPECT_EQ(buf, data);
            if (ouput_debug) {
                std::cout << "index:" << i << " data_len:" << total_len << " 4 queue dump:" << queue->dump()
                          << std::endl;
            }

            queue->readOver(static_cast<xncc::types::queue_size_t>(head->len) + THREAD_MSG_HEAD_LEN);
            if (ouput_debug) {
                std::cout << "index:" << i << " data_len:" << total_len << " 5 queue dump:" << queue->dump()
                          << std::endl;
            }
        }
        else {
            std::cout << "read second nullptr i = " << i << " len " << total_len << " dump:" << queue->dump()
                      << std::endl;
        }
    }
#endif
}

TEST_F(FIFOSafeQueueTestSuite, read_nullptr)
{
#ifdef DEBUG_PROD

    for (int i = 0; i < 1000000; i++) {
        char* readBuf = queue->read(THREAD_MSG_HEAD_LEN);
        if (unlikely(readBuf != nullptr)) {
            std::cout << "error nullptr i = " << i << " len " << THREAD_MSG_HEAD_LEN << " dump:" << queue->dump()
                      << std::endl;
        }

        EXPECT_TRUE(readBuf == nullptr);
    }
#endif
}

TEST_F(FIFOSafeQueueTestSuite, write_more_than_once)
{
#ifdef DEBUG_PROD

    xncc::tools::xncc_random<xncc::types::queue_size_t> random(24, 1000);
    xncc::tools::xncc_random<std::string>               randomStr;
    bool                                                ouput_debug = false;
    const int                                           one_times   = 10;
    for (int i = 0; i < 1000000; i++) {
        const auto len       = random.get_value();
        const auto buf       = randomStr.get_value(int(len));
        const auto total_len = THREAD_MSG_HEAD_LEN + len;
        for (int j = 0; j < one_times; j++) {
            char                          buffer[THREAD_MSG_HEAD_LEN + 1001] = {0};
            xncc::msg::MsgHeaderInThread* hptr = reinterpret_cast<xncc::msg::MsgHeaderInThread*>(buffer);
            hptr->len                          = static_cast<uint32_t>(len);
            hptr->id                           = 1000;
            hptr->connId                       = 1;
            hptr->tid                          = 1;
            memcpy(buffer + THREAD_MSG_HEAD_LEN, buf.c_str(), static_cast<uint32_t>(len));
            const xncc::types::queue_size_t total_len = THREAD_MSG_HEAD_LEN + len;
            if (ouput_debug) {
                std::cout << "index:" << i << " data_len:" << total_len << " 1 queue dump:" << queue->dump()
                          << std::endl;
            }
            queue->write(buffer, total_len);
            queue->writeOverNotNotify(total_len);
        }

        for (int j = 0; j < one_times; j++) {
            char* readBuf = queue->read(THREAD_MSG_HEAD_LEN);
            if (unlikely(readBuf == nullptr)) {
                std::cout << "nullptr i = " << i << " len " << total_len << " dump:" << queue->dump() << std::endl;
            }
            xncc::msg::MsgHeaderInThread* head = reinterpret_cast<xncc::msg::MsgHeaderInThread*>(readBuf);
            if (head->len != len) {
                std::cout << "read error i = " << i << " len " << total_len << " dump:" << queue->dump() << std::endl;
            }
            EXPECT_EQ(head->len, len);
            EXPECT_EQ(head->id, 1000);
            EXPECT_EQ(head->connId, 1);
            EXPECT_EQ(head->tid, 1);

            if (ouput_debug) {
                std::cout << "index:" << i << " data_len:" << total_len << " 3 queue dump:" << queue->dump()
                          << std::endl;
            }

            char* dataBuf = queue->read(static_cast<xncc::types::queue_size_t>(head->len) + THREAD_MSG_HEAD_LEN);
            EXPECT_TRUE(dataBuf != nullptr);
            if (likely(dataBuf)) {
                const auto data = std::string(dataBuf + THREAD_MSG_HEAD_LEN, head->len);
                EXPECT_EQ(buf, data);
                if (ouput_debug) {
                    std::cout << "index:" << i << " data_len:" << total_len << " 4 queue dump:" << queue->dump()
                              << std::endl;
                }

                queue->readOver(static_cast<xncc::types::queue_size_t>(head->len) + THREAD_MSG_HEAD_LEN);
                if (ouput_debug) {
                    std::cout << "index:" << i << " data_len:" << total_len << " 5 queue dump:" << queue->dump()
                              << std::endl;
                }
            }
            else {
                std::cout << "read second nullptr i = " << i << " len " << total_len << " dump:" << queue->dump()
                          << std::endl;
            }
        }
    }
#endif
}

TEST_F(FIFOSafeQueueTestSuite, one_thread_write_one_thread_read)
{
#ifdef DEBUG_PROD

    xncc::tools::xncc_random<xncc::types::queue_size_t> random(24, 1000);
    bool                                                ouput_debug = false;
    FILE*                                               fp          = fopen("/dev/shm/queue_dump", "w");

    auto func = [&random](TestQueue* queue, bool ouput_debug, FILE* fp) {
        for (int i = 0; i < 1000000; i++) {
            const auto len  = random.get_value();
            int        wlen = static_cast<int>(len);
            fwrite(&i, sizeof(i), 1, fp);
            fwrite(&wlen, sizeof(wlen), 1, fp);
            char                          buffer[THREAD_MSG_HEAD_LEN + 1001] = {0};
            xncc::msg::MsgHeaderInThread* hptr = reinterpret_cast<xncc::msg::MsgHeaderInThread*>(buffer);
            hptr->len                          = static_cast<uint32_t>(len);
            hptr->id                           = 1000;
            hptr->connId                       = 1;
            hptr->tid                          = 1;
            memcpy(buffer + THREAD_MSG_HEAD_LEN, &i, sizeof(int));
            const xncc::types::queue_size_t total_len = THREAD_MSG_HEAD_LEN + len;
            if (ouput_debug) {
                std::cout << "index:" << i << " data_len:" << total_len << " 1 queue dump:" << queue->dump()
                          << std::endl;
            }
            queue->write(buffer, total_len);
            queue->writeOverNotNotify(total_len);
        }

        fclose(fp);
    };

    std::thread t1(func, queue, ouput_debug, fp);

    int i = 0;
    while (true) {
        char* readBuf = queue->read(THREAD_MSG_HEAD_LEN);
        if (unlikely(readBuf == nullptr)) {
            // std::cout << "nullptr i = " << i << " len " << THREAD_MSG_HEAD_LEN << " dump:" << queue->dump()
            //           << std::endl;
            continue;
        }
        xncc::msg::MsgHeaderInThread* head = reinterpret_cast<xncc::msg::MsgHeaderInThread*>(readBuf);
        // if (head->len != len) {
        //     std::cout << "read error i = " << i << " len " << total_len << " dump:" << queue->dump() << std::endl;
        // }
        // EXPECT_EQ(head->len, len);
        EXPECT_EQ(head->id, 1000);
        EXPECT_EQ(head->connId, 1);
        EXPECT_EQ(head->tid, 1);

        auto total_len = head->len + THREAD_MSG_HEAD_LEN;
        ++i;
        if (ouput_debug) {
            std::cout << "index:" << i << " data_len:" << total_len << " 3 queue dump:" << queue->dump() << std::endl;
        }

        char* dataBuf = queue->read(static_cast<xncc::types::queue_size_t>(head->len) + THREAD_MSG_HEAD_LEN);
        EXPECT_TRUE(dataBuf != nullptr);
        if (likely(dataBuf)) {
            if (ouput_debug) {
                std::cout << "index:" << i << " data_len:" << total_len << " 4 queue dump:" << queue->dump()
                          << std::endl;
            }
            queue->readOver(static_cast<xncc::types::queue_size_t>(head->len) + THREAD_MSG_HEAD_LEN);
            if (ouput_debug) {
                std::cout << "index:" << i << " data_len:" << total_len << " 5 queue dump:" << queue->dump()
                          << std::endl;
            }
        }
        else {
            std::cout << "read second nullptr i = " << i << " len " << total_len << " dump:" << queue->dump()
                      << std::endl;
        }
        if (i % 100000 == 0) {
            std::cout << "index:" << i << " data_len:" << total_len << " 6 queue dump:" << queue->dump() << std::endl;
        }
        if (i == 1000000) {
            break;
        }
    }

    EXPECT_TRUE(i == 1000000);
    t1.join();
#endif
}

TEST_F(FIFOSafeQueueTestSuite, one_thread_write_one_thread_read_read_file)
{
#ifdef DEBUG_PROD

    bool  ouput_debug = false;
    FILE* fp          = fopen("/dev/shm/queue_dump", "r");

    auto func = [](TestQueue* queue, bool ouput_debug, FILE* fp) {
        for (int i = 0; i < 1000000; i++) {
            int  tmpid;
            auto ret = fread(&tmpid, sizeof(tmpid), 1, fp);
            if (ret != 1) {
                std::cout << "read error i = " << i << " len " << tmpid << " dump:" << queue->dump() << std::endl;
            }
            int wlen;
            ret = fread(&wlen, sizeof(wlen), 1, fp);
            if (ret != 1) {
                std::cout << "read error i = " << i << " len " << tmpid << " dump:" << queue->dump() << std::endl;
            }
            char                          buffer[THREAD_MSG_HEAD_LEN + 1001] = {0};
            xncc::msg::MsgHeaderInThread* hptr = reinterpret_cast<xncc::msg::MsgHeaderInThread*>(buffer);
            hptr->len                          = static_cast<uint32_t>(wlen);
            hptr->id                           = 1000;
            hptr->connId                       = 1;
            hptr->tid                          = 1;
            memcpy(buffer + THREAD_MSG_HEAD_LEN, &i, sizeof(int));
            const xncc::types::queue_size_t total_len =
                static_cast<xncc::types::queue_size_t>(THREAD_MSG_HEAD_LEN + static_cast<uint32_t>(wlen));
            if (ouput_debug) {
                std::cout << "index:" << i << " data_len:" << total_len << " 1 queue dump:" << queue->dump()
                          << std::endl;
            }
            queue->write(buffer, total_len);
            queue->writeOverNotNotify(total_len);
        }

        fclose(fp);
    };

    std::thread t1(func, queue, ouput_debug, fp);

    FILE* fpr = fopen("/dev/shm/read_result", "w");

    int i = 0;
    while (true) {
        char* readBuf = queue->read(THREAD_MSG_HEAD_LEN);
        if (unlikely(readBuf == nullptr)) {
            // std::cout << "nullptr i = " << i << " len " << THREAD_MSG_HEAD_LEN << " dump:" << queue->dump()
            //           << std::endl;
            continue;
        }
        xncc::msg::MsgHeaderInThread* head = reinterpret_cast<xncc::msg::MsgHeaderInThread*>(readBuf);
        // if (head->len != len) {
        //     std::cout << "read error i = " << i << " len " << total_len << " dump:" << queue->dump() << std::endl;
        // }
        // EXPECT_EQ(head->len, len);
        EXPECT_EQ(head->id, 1000);
        EXPECT_EQ(head->connId, 1);
        EXPECT_EQ(head->tid, 1);

        auto total_len = head->len + THREAD_MSG_HEAD_LEN;
        ++i;
        char* dataBuf = queue->read(total_len);
        EXPECT_TRUE(dataBuf != nullptr);
        if (likely(dataBuf)) {
            const int data = *(reinterpret_cast<int*>(dataBuf + THREAD_MSG_HEAD_LEN));
            fwrite(&data, sizeof(int), 1, fpr);
            const int wlen = static_cast<int>(head->len);
            fwrite(&wlen, sizeof(int), 1, fpr);
            queue->readOver(total_len);
        }
        else {
            std::cout << "read second nullptr i = " << i << " len " << total_len << " dump:" << queue->dump()
                      << std::endl;
        }
        if (i % 100000 == 0) {
            std::cout << "index:" << i << " data_len:" << total_len << " 6 queue dump:" << queue->dump() << std::endl;
        }
        if (i == 1000000) {
            break;
        }
    }

    EXPECT_TRUE(i == 1000000);
    t1.join();
#endif
}